/**
 * LY.com Inc.
 * Copyright (c) 2004-2021 All Rights Reserved.
 */
package quorum;

import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Quorum算法，暂时不支持节点扩容
 * @author bixiao.wang
 * @version $Id: AbstractQuorum, v 0.1 2021/9/18 14:37 bixiao.wang Exp $
 */
public abstract class AbstractQuorum {

    private List<Node> activeNodeList = new LinkedList<>();
    private List<Node> inactiveNodeList = new LinkedList<>();
    private List<Node> nodeList;

    /**
     * 协调数据一致性的时候，是否读取具体的数据，否的话只读取版本信息，
     * 当协调完成后再发请求去读数据
     */
    private boolean readDataWhenJudge;

    private int n;
    private int w;
    private int r;
    private ExecutorService writeTaskService = Executors.newFixedThreadPool(5);
    private ExecutorService readTaskService = Executors.newFixedThreadPool(5);
    private AtomicInteger readIdx = new AtomicInteger();


    public AbstractQuorum(List<Node> nodeList, int w, int r, boolean readDataWhenJudge) {
        int n = nodeList.size();
        if (w + r <= n) {
            throw new IllegalArgumentException("w+r必须大于n");
        }
        if (n/2 + 1 > w) {
            throw new IllegalArgumentException("w必须大于n/2+1");
        }
        this.nodeList = nodeList;
        this.n = n;
        this.w = w;
        this.r = r;
        this.readDataWhenJudge = readDataWhenJudge;
    }

    public boolean start() {
        final List<Node> nodeList = this.nodeList;
        for (Node node : nodeList) {
            try {
                boolean success = init(node);
                if (success) {
                    activeNodeList.add(node);
                } else {
                    inactiveNodeList.add(node);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        int activeNum = activeNodeList.size();
        return activeNum >= w && activeNum >= r;
    }

    /**
     * 写数据
     * @param data
     * @param timeout 超时时间，毫秒
     * @return
     */
    public boolean write(String key, Data data, long timeout) {
        if (activeNodeList.size() < w) {
            return false;
        }
        DataWrapper dataWrapper = new DataWrapper(System.currentTimeMillis(), data);

        CountDownLatch latch = new CountDownLatch(w);
        AtomicInteger success = new AtomicInteger();
        Thread current = Thread.currentThread();
        for (Node node : activeNodeList) {
            writeTaskService.execute(() -> {
                boolean result = doWrite(key, node, dataWrapper);
                if (result) {
                    int successTask = success.incrementAndGet();
                    if (successTask >= w) {
                        current.interrupt();
                        return;
                    }
                }
                latch.countDown();
            });
        }
        try {
            latch.await(timeout, TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
        }
        return success.get() >= w;
    }

    public Data read(String key, Long readTimeoutPerNode) throws Exception {
        if (activeNodeList.size() < r) {
            throw new Exception("服务不可用");
        }

        List<Node> nodeList = new LinkedList<>(activeNodeList);
        Collections.shuffle(nodeList);
        List<DataWrapper> dataWrapperList = new LinkedList<>();
        List<Node> succeseeNode = new LinkedList<>();
        int success = 0, rest = nodeList.size();
        long maxVersion = -1;
        while (success < r && nodeList.size() != 0) {
            int addNum = Math.min(r - success, nodeList.size());
            List<Future<DataWrapper>> resultPart = new LinkedList<>();
            for (int i = 0; i < addNum; i++) {
                Future<DataWrapper> resultFuture = readTaskService.submit(() -> {
                    Node node = nodeList.remove(0);
                    DataWrapper dataWrapper = null;
                    if (readDataWhenJudge) {
                        dataWrapper = doReadData(node, key);
                    } else {
                        dataWrapper = doReadMeta(node, key);
                    }
                    if (dataWrapper != null) {
                        dataWrapper.setNode(node);
                    }
                    return dataWrapper;
                });
                resultPart.add(resultFuture);
            }
            for (Future<DataWrapper> dataWrapperFuture : resultPart) {
                try {
                    DataWrapper dataWrapper = dataWrapperFuture.get(readTimeoutPerNode, TimeUnit.MILLISECONDS);
                    if (dataWrapper != null) {
                        if (dataWrapper.getVersion() > maxVersion) {
                            maxVersion = dataWrapper.getVersion();
                            success = 1;
                            dataWrapperList.clear();
                            succeseeNode.clear();
                            dataWrapperList.add(dataWrapper);
                        } else if (dataWrapper.getVersion() == maxVersion) {
                            success++;
                            dataWrapperList.add(dataWrapper);
                        }
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                } catch (TimeoutException e) {
                    e.printStackTrace();
                }
            }
            rest -= addNum;
        }
        // 查询不到数据，失败
        if (dataWrapperList.size() == 0) {
            return null;
        }
        if (dataWrapperList.size() >= r) {
            // 查询成功
            if (this.readDataWhenJudge) {
                return dataWrapperList.get(0).getData();
            } else {
                for (DataWrapper dataWrapper : dataWrapperList) {
                    DataWrapper dataWrapperResult = doReadData(dataWrapper.getNode(), key);
                    if (dataWrapperResult != null && dataWrapperResult.getData() != null) {
                        return dataWrapperResult.getData();
                    }
                }
            }
        }
        // 最高的版本小于r，说明写失败，删除无效版本，重新查询
        for (DataWrapper dataWrapper : dataWrapperList) {
            doDeleteData(dataWrapper.getNode(), key, dataWrapper.getVersion());
        }
        return read(key, readTimeoutPerNode);
    }

    public void delete(String key) {
        for (Node node : activeNodeList) {
            doDeleteData(node, key);
        }
    }

    public void shutdown() {
        for (Node node : activeNodeList) {
            try {
                destroy(node);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 初始化一个与节点的连接
     * @param node
     * @return
     */
    public abstract boolean init(Node node);

    public abstract boolean destroy(Node node);

    /**
     * 向一个节点写数据
     * @param node
     * @return
     */
    public abstract boolean doWrite(String key, Node node, DataWrapper dataWrapper);

    /**
     * 只读取版本信息
     * @param node
     * @param key
     * @return
     */
    public abstract DataWrapper doReadMeta(Node node, String key);

    /**
     * 读取版本和数据
     * @param node
     * @param key
     * @return
     */
    public abstract DataWrapper doReadData(Node node, String key);

    public abstract boolean doDeleteData(Node node, String key, Long version);

    public abstract boolean doDeleteData(Node node, String key);
}
